package com.htiiot.status.write;

import com.htiiot.status.config.Configs;
import com.htiiot.status.hdfs.HDFSWriter;
import com.htiiot.status.hdfs.HTWriter;
import com.htiiot.status.hdfs.LocalWriter;
import com.htiiot.status.kafka.Listener;
import org.apache.logging.log4j.LogManager;

import java.io.IOException;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

public class WriterCache {
	private static org.apache.logging.log4j.Logger logger = LogManager.getLogger(WriterCache.class);

	private static Map<String, HTWriter> writerMap = new ConcurrentHashMap<String, HTWriter>();

	/**
	 * 从缓存中获取Writer
	 * 
	 * @param tID
	 * @param host
	 * @return
	 */
	public static HTWriter getWriter(long tID, String host) {
		String key = tID + "_host_" + host;
		HTWriter writer = writerMap.get(key);
		if (writer == null) {
			synchronized (Listener.class) {
				if (writer == null) {
					writer = createWriter( tID, host );
				}
			}
		}
		return writer;
	}

	/**
	 * 创建Writer并放入缓存writerMap中
	 * 
	 * @param tID
	 * @param host
	 * @return
	 */
	public static HTWriter createWriter(long tID,String host) {
		String key = tID + "_host_" + host;
		HTWriter writer = null;
		try {
			if ("LOCAL".equals( Configs.WriteConfig.WRITE_TYPE.toUpperCase())) {
				writer = new LocalWriter(tID, host, System.currentTimeMillis());
			} else {
				writer = new HDFSWriter(tID, host, System.currentTimeMillis());
			}
			writerMap.put(key, writer);
		} catch (IOException e) {
			logger.error(e);
		}
		return writer;
	}

	/**
	 * 更新Writer，对每台设备每个线程的writer进行更新
	 * 
	 * @param tID
	 * @param host
	 * @return
	 */
	public static HTWriter updateWriter(long tID, String host) {
		HTWriter writer = createWriter(tID, host);
		return writer;
	}

	/**
	 * 更新所有writer
	 */
	public static void updateWriter() {
		logger.info("更新所有writer");
		Listener.writeLock.lock();
		Set<String> keySet = new HashSet<String>(writerMap.keySet());
		try {
			for (String key : keySet) {
				HTWriter writer = writerMap.get( key );
				long tID = writer.getTid();
				String host = writer.getHost();
				updateWriter( tID, host );
				writer.close();
			}
		}finally {
			Listener.writeLock.unlock();
		}
	}
	/**
	 * 关闭所有文件流
	 * @return
	 */
	public static void closeWriter(){
		logger.info("开始关闭所有writer");
		Set<String> keySet = new HashSet<String>(writerMap.keySet());
		for (String key : keySet) {
			HTWriter writer = writerMap.get(key);
			writer.close();
		}
		logger.info("关闭文件流结束，共关闭：" + writerMap.size() + " 个文件流。");
	}

	public static int getMapSize() {
		return writerMap.size();
	}

}
